package org.onion.pattern.event.outbox;

import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.onion.pattern.event.io.EventDeserializer;
import org.onion.pattern.event.io.codec.JacksonEventCodec;
import org.onion.pattern.event.io.codec.ProtobufEventCodec;
import org.onion.pattern.event.outbox.converter.DefaultOutboxEventConverter;
import org.onion.pattern.event.outbox.data.OutboxEventInfo;
import org.onion.pattern.event.outbox.executor.OutboxStorageDeliveryExecutor;
import org.onion.pattern.event.outbox.storage.MemoryOutboxEventStorage;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.stream.Collectors;

@Getter
@Setter
@Slf4j
@Deprecated
public class DefaultEventOutbox implements EventOutbox {

    private OutboxEventStorage         storage            = new MemoryOutboxEventStorage();
    private List<OutboxEventConverter> converters         = new ArrayList<>();
    private List<EventDeserializer> eventDeserializers = new ArrayList<>();
    private OutboxDeliveryExecutor  deliveryExecutor;

    public void init() {
        if (deliveryExecutor == null) {
            deliveryExecutor = new OutboxStorageDeliveryExecutor(this.storage);
        }
    }

    @Override
    public void flush() {
        deliveryExecutor.flush();
    }

    @Override
    public void shutdown() {
        deliveryExecutor.shutdown();
    }

    @Override
    public void start() {
        deliveryExecutor.start();
    }

    @Override
    public void delivery(Object event) {
        //TODO 创建多个渠道的事件 2025-08-27 23:15  何挺强
        OutboxEventInfo eventInfo = toOutboxInfo(event);
        this.storage.insert(eventInfo);
    }

    @Override
    public void delivery(List<Object> events) {
        List<OutboxEventInfo> collect = events.stream()
                .map(this::toOutboxInfo)
                .collect(Collectors.toList());
        this.storage.batchInsert(collect);
    }

    @Override
    public Optional<Object> retrieval(String key) {
        Optional<OutboxEventInfo> byEventKey = this.storage.findByEventKey(key);
        if (!byEventKey.isPresent()) {
            return Optional.empty();
        }
        Object object = toObject(byEventKey.get());
        return Optional.of(object);
    }


    @SneakyThrows
    private Object toObject(OutboxEventInfo eventInfo) {
        Class<?> clazz = Class.forName(eventInfo.getDomainType());
        byte[] payload = eventInfo.getPayload();
        String serializer = eventInfo.getSerializer();

        if (CollectionUtils.isNotEmpty(eventDeserializers)) {
            for (EventDeserializer eventDeserializer : eventDeserializers) {
                if (eventDeserializer.type().equalsIgnoreCase(serializer)) {
                    return eventDeserializer.deserialize(payload, clazz);
                }
            }
        }

        if (JacksonEventCodec.TYPE.equalsIgnoreCase(serializer)) {
            return JacksonEventCodec.INSTANCE.deserialize(payload, clazz);
        }

        if (ProtobufEventCodec.TYPE.equalsIgnoreCase(serializer)) {
            return ProtobufEventCodec.INSTANCE.deserialize(payload, clazz);
        }

        throw new IllegalArgumentException("无法解析对象 " + eventInfo.getEventKey() + "类型: " + eventInfo.getDomainType());
    }

    private OutboxEventInfo toOutboxInfo(Object event) {
        if (CollectionUtils.isNotEmpty(converters)) {
            for (OutboxEventConverter converter : converters) {
                if (converter.support(event)) {
                    return converter.convert(event);
                }
            }
        }
        return DefaultOutboxEventConverter.INSTANCE.convert(event);
    }

}
